package com.ddone.listener;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.stereotype.Component;

/**
 * @author ddone
 * @date 2024/5/13-23:20
 * // 就是一个普通的类，最好是service业务类
 * // 不需要web请求
 * // @KafkaListener 加了这个注解之后，for(;;) white(true)
 */
@Component
public class KafkaDemoListenerOffset {

    @KafkaListener(
            topicPartitions = {
                    @TopicPartition(
                            topic = "spring-test-partition-topic",
                            //消费分区2所有的消息
                            partitions = {"2"},
                            //指定消费的分区，并配置消费该分期的起始offset,注意partitions中配置的主题不能重复
                            partitionOffsets = {
                                    //分区0 offset从3开始消费
                                    @PartitionOffset(
                                            partition = "0",
                                            initialOffset = "3",
                                            relativeToCurrent="true"
                                    ),
                                    //分区1 offset从0开始消费
                                    @PartitionOffset(
                                            partition = "1",
                                            initialOffset = "0",
                                            relativeToCurrent="true"
                                    ),
                            }
                    )
            },
            concurrency = "3"
    )
    public void consumeByPartitionOffsets(ConsumerRecord<String, String> record) {
        System.out.println("consumeByPartitionOffsets");
        System.out.printf(
                "主题 = %s,分区 = %d, 偏移量 = %d, key = %s, 内容 = %s,时间戳 = %d%n",
                record.topic(),
                record.partition(),
                record.offset(),
                record.key(),
                record.value(),
                record.timestamp()
        );
    }

}
